use std::mem;
use std::ops::Deref;
use std::ops::DerefMut;

use windows::core::Error as WinError;
use windows::Win32::System::Threading::INFINITE;
use windows::Win32::System::IO as Win32IO;
use windows::core::HSTRING;
use windows::Win32::Foundation;


use super::async_data::Overlapped;
use super::nodebug::NoDebug;
use super::ipc_receiver::OsIpcReceiver;
use super::share_memory::OsIpcSharedMemory;
use super::utils;
use crate::win32_trace;

use super::{aliased_cell::AliasedCell, async_data::AsyncData, enums::{BlockingMode, WinIpcError}, ipc_bridge::OsOpaqueIpcChannel, win_handle::WinHandle};


pub(super) struct MessageHeader {
    pub(super) data_len: u32,
    pub(super) oob_len: u32,
}

impl MessageHeader {
    pub(super) fn total_message_bytes_needed(&self) -> usize {
        mem::size_of::<MessageHeader>() + self.data_len as usize + self.oob_len as usize
    }
}

pub(super) struct Message<'data> {
    data_len: usize,
    oob_len: usize,
    bytes: &'data [u8],
}

impl<'data> Message<'data> {
    fn from_bytes(bytes: &'data [u8]) -> Option<Message> {
        if bytes.len() < mem::size_of::<MessageHeader>() {
            return None;
        }

        unsafe {
            let ref header = *(bytes.as_ptr() as *const MessageHeader);
            if bytes.len() < header.total_message_bytes_needed() {
                return None;
            }

            Some(Message {
                data_len: header.data_len as usize,
                oob_len: header.oob_len as usize,
                bytes: &bytes[0..header.total_message_bytes_needed()],
            })
        }
    }

    fn data(&self) -> &[u8] {
        &self.bytes
            [mem::size_of::<MessageHeader>()..(mem::size_of::<MessageHeader>() + self.data_len)]
    }

    fn oob_bytes(&self) -> &[u8] {
        &self.bytes[(mem::size_of::<MessageHeader>() + self.data_len)..]
    }

    fn oob_data(&self) -> Option<OutOfBandMessage> {
        if self.oob_len > 0 {
            let oob = bincode::deserialize::<OutOfBandMessage>(self.oob_bytes())
                .expect("Failed to deserialize OOB data");
            if oob.target_process_id != *utils::CURRENT_PROCESS_ID {
                panic!("Windows IPC channel received handles intended for pid {}, but this is pid {}. \
                       This likely happened because a receiver was transferred while it had outstanding data \
                       that contained a channel or shared memory in its pipe. \
                       This isn't supported in the Windows implementation.",
                       oob.target_process_id, *utils::CURRENT_PROCESS_ID);
            }
            Some(oob)
        } else {
            None
        }
    }

    fn size(&self) -> usize {
        mem::size_of::<MessageHeader>() + self.data_len + self.oob_len
    }
}



/// If we have any channel handles or shmem segments, then we'll send an
/// OutOfBandMessage after the data message.
///
/// This includes the receiver's process ID, which the receiver checks to
/// make sure that the message was originally sent to it, and was not sitting
/// in another channel's buffer when that channel got transferred to another
/// process.  On Windows, we duplicate handles on the sender side to a specific
/// receiver.  If the wrong receiver gets it, those handles are not valid.
///
/// TODO(vlad): We could attempt to recover from the above situation by
/// duplicating from the intended target process to ourselves (the receiver).
/// That would only work if the intended process a) still exists; b) can be
/// opened by the receiver with handle dup privileges.  Another approach
/// could be to use a separate dedicated process intended purely for handle
/// passing, though that process would need to be global to any processes
/// amongst which you want to share channels or connect one-shot servers to.
/// There may be a system process that we could use for this purpose, but
/// I haven't found one -- and in the system process case, we'd need to ensure
/// that we don't leak the handles (e.g. dup a handle to the system process,
/// and then everything dies -- we don't want those resources to be leaked).
#[derive(Debug)]
pub(super) struct OutOfBandMessage {
    pub(super) target_process_id: u32,
    pub(super) channel_handles: Vec<isize>,
    pub(super) shmem_handles: Vec<(isize, u64)>, // handle and size
    pub(super) big_data_receiver_handle: Option<(isize, u64)>, // handle and size
}

impl OutOfBandMessage {
    pub(super) fn new(target_id: u32) -> OutOfBandMessage {
        OutOfBandMessage {
            target_process_id: target_id,
            channel_handles: vec![],
            shmem_handles: vec![],
            big_data_receiver_handle: None,
        }
    }

    pub(super) fn needs_to_be_sent(&self) -> bool {
        !self.channel_handles.is_empty()
            || !self.shmem_handles.is_empty()
            || self.big_data_receiver_handle.is_some()
    }
}

impl serde::Serialize for OutOfBandMessage {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        ((
            self.target_process_id,
            &self.channel_handles,
            &self.shmem_handles,
            &self.big_data_receiver_handle,
        ))
            .serialize(serializer)
    }
}

impl<'de> serde::Deserialize<'de> for OutOfBandMessage {
    fn deserialize<D>(deserializer: D) -> Result<OutOfBandMessage, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let (target_process_id, channel_handles, shmem_handles, big_data_receiver_handle) =
            serde::Deserialize::deserialize(deserializer)?;
        Ok(OutOfBandMessage {
            target_process_id: target_process_id,
            channel_handles: channel_handles,
            shmem_handles: shmem_handles,
            big_data_receiver_handle: big_data_receiver_handle,
        })
    }
}



/// Main object keeping track of a receive handle and its associated state.
///
/// Implements blocking/nonblocking reads of messages from the handle.
#[derive(Debug)]
pub(super) struct MessageReader {
    /// The pipe read handle.
    ///
    /// Note: this is only set while no async read operation
    /// is currently in progress with the kernel.
    /// When an async read is in progress,
    /// it is moved into the `async` sub-structure (see below)
    /// along with the other fields used for the async operation,
    /// to make sure they all stay in sync,
    /// and nothing else can meddle with the the pipe
    /// until the operation is completed.
    pub(super) handle: WinHandle,

    /// Buffer for outstanding data, that has been received but not yet processed.
    ///
    /// Note: just like `handle` above,
    /// this is only set while no async read is in progress.
    /// When an async read is in progress,
    /// the receive buffer is aliased by the kernel;
    /// so we need to temporarily move it into an `AliasedCell`,
    /// thus making it inaccessible from safe code --
    /// see `async` below.
    /// We only move it back once the kernel signals completion of the async read.
    pub(super) read_buf: Vec<u8>,

    /// Data used by the kernel during an async read operation.
    ///
    /// Note: Since this field only has a value
    /// when an async read operation is in progress
    /// (i.e. has been issued to the system, and not completed yet),
    /// this also serves as an indicator of the latter.
    ///
    /// WARNING: As the kernel holds mutable aliases of this data
    /// while an async read is in progress,
    /// it is crucial that it is never accessed in user space
    /// from the moment we issue an async read in `start_read()`,
    /// until the moment we process the event
    /// signalling completion of the async read in `notify_completion()`.
    ///
    /// Since Rust's type system is not aware of the kernel aliases,
    /// the compiler cannot guarantee exclusive access the way it normally would,
    /// i.e. any access to this value is inherently unsafe!
    /// We thus wrap it in an `AliasedCell`,
    /// making sure the data is only accessible from code marked `unsafe`;
    /// and only move it out when the kernel signals that the async read is done.
    pub(super) r#async: Option<AliasedCell<AsyncData>>,

    /// Token identifying the reader/receiver within an `OsIpcReceiverSet`.
    ///
    /// This is returned to callers of `OsIpcReceiverSet.add()` and `OsIpcReceiverSet.select()`.
    ///
    /// `None` if this `MessageReader` is not part of any set.
    pub(super) entry_id: Option<u64>,
}


impl Drop for MessageReader {
    fn drop(&mut self) {
        // Before dropping the `ov` structure and read buffer,
        // make sure the kernel won't do any more async updates to them!
        self.cancel_io();
    }
}

impl MessageReader {
    pub(super) fn new(handle: WinHandle) -> MessageReader {
        MessageReader {
            handle,
            read_buf: Vec::new(),
            r#async: None,
            entry_id: None,
        }
    }

    pub(super) fn take(&mut self) -> MessageReader {
        // This is currently somewhat inefficient,
        // because of the initialisation of things that won't be used.
        // Moving the data items of `MessageReader` into an enum will fix this,
        // as that way we will be able to just define a data-less `Invalid` case.
        mem::replace(self, MessageReader::new(WinHandle::invalid()))
    }

    /// Request the kernel to cancel a pending async I/O operation on this reader.
    ///
    /// Note that this only schedules the cancel request;
    /// but doesn't guarantee that the operation is done
    /// (and the buffers are no longer used by the kernel)
    /// before this method returns.
    ///
    /// A caller that wants to ensure the operation is really done,
    /// will need to wait using `fetch_async_result()`.
    /// (Or `fetch_iocp_result()` for readers in a set.)
    ///
    /// The only exception is if the kernel indicates
    /// that no operation was actually outstanding at this point.
    /// In that case, the `async` data is released immediately;
    /// and the caller should not attempt waiting for completion.
    pub(super) fn issue_async_cancel(&mut self) {
        unsafe {
            let status = Win32IO::CancelIoEx(
                self.r#async.as_ref().unwrap().alias().handle.as_raw(),
                self.r#async
                    .as_ref()
                    .map(|a| std::ptr::addr_of!(a.alias().ov.0.deref().0)),
            );

            if status == false {
                // A cancel operation is not expected to fail.
                // If it does, callers are not prepared for that -- so we have to bail.
                //
                // Note that we should never ignore a failed cancel,
                // since that would affect further operations;
                // and the caller definitely must not free the aliased data in that case!
                //
                // Sometimes `CancelIoEx()` fails with `ERROR_NOT_FOUND` though,
                // meaning there is actually no async operation outstanding at this point.
                // (Specifically, this is triggered by the `receiver_set_big_data()` test.)
                // Not sure why that happens -- but I *think* it should be benign...
                //
                // In that case, we can safely free the async data right now;
                // and the caller should not attempt to wait for completion.
                assert!(WinError::from_win32().code() == Foundation::ERROR_NOT_FOUND.to_hresult());

                let async_data = self.r#async.take().unwrap().into_inner();
                self.handle = async_data.handle;
                self.read_buf = async_data.buf;
            }
        }
    }

    pub(super) fn cancel_io(&mut self) {
        if self.r#async.is_some() {
            // This doesn't work for readers in a receiver set.
            // (`fetch_async_result()` would hang indefinitely.)
            // Receiver sets have to handle cancellation specially,
            // and make sure they always do that *before* dropping readers.
            assert!(self.entry_id.is_none());

            self.issue_async_cancel();

            // If there is an operation still in flight, wait for it to complete.
            //
            // This will usually fail with `ERROR_OPERATION_ABORTED`;
            // but it could also return success, or some other error,
            // if the operation actually completed in the mean time.
            // We don't really care either way --
            // we just want to be certain there is no operation in flight any more.
            if self.r#async.is_some() {
                let _ = self.fetch_async_result(BlockingMode::Blocking);
            }
        }
    }

    /// Kick off an asynchronous read.
    ///
    /// When an async read is started successfully,
    /// the receive buffer is moved out of `read_buf`
    /// into the `AliasedCell<>` in `async`,
    /// thus making it inaccessible from safe code;
    /// it will only be moved back in `notify_completion()`.
    /// (See documentation of the `read_buf` and `async` fields.)
    pub(super) fn start_read(&mut self) -> Result<(), WinIpcError> {
        // Nothing needs to be done if an async read operation is already in progress.
        if self.r#async.is_some() {
            return Ok(());
        }

        win32_trace!("[$ {:?}] start_read", self.handle);

        if self.read_buf.len() == self.read_buf.capacity() {
            self.read_buf.reserve(utils::PIPE_BUFFER_SIZE);
        }

        unsafe {
            // Temporarily extend the vector to span its entire capacity,
            // so we can safely sub-slice it for the actual read.
            let buf_len = self.read_buf.len();
            let buf_cap = self.read_buf.capacity();
            self.read_buf.set_len(buf_cap);

            // issue the read to the buffer, at the current length offset
            self.r#async = Some(AliasedCell::new(AsyncData {
                handle: self.handle.take(),
                ov: NoDebug(Box::new({
                    let mut overlapped: Win32IO::OVERLAPPED = mem::zeroed();
                    // Create a manually reset event. The documentation for GetOverlappedResultEx
                    // states you must do this in the remarks section.
                    overlapped.hEvent = windows::Win32::System::Threading::CreateEventA(None, true, false, None)?;
                    Overlapped::new(overlapped)
                })),
                buf: mem::replace(&mut self.read_buf, vec![]),
            }));
            let ok = {
                let async_data = self.r#async.as_mut().unwrap().alias_mut();
                let remaining_buf = &mut async_data.buf[buf_len..];
                windows::Win32::Storage::FileSystem::ReadFile(
                    async_data.handle.as_raw(),
                    Some(remaining_buf.as_mut_ptr() as _),
                    remaining_buf.len() as u32,
                    None,
                    Some(&mut ***async_data.ov.deref_mut()),
                )
            };

            // Reset the vector to only expose the already filled part.
            //
            // This means that the async read
            // will actually fill memory beyond the exposed part of the vector.
            // While this use of a vector is officially sanctioned for such cases,
            // it still feel rather icky to me...
            //
            // On the other hand, this way we make sure
            // the buffer never appears to have more valid data
            // than what is actually present,
            // which could pose a potential danger in its own right.
            // Also, it avoids the need to keep a separate state variable --
            // which would bear some risk of getting out of sync.
            self.r#async
                .as_mut()
                .unwrap()
                .alias_mut()
                .buf
                .set_len(buf_len);

            let result = if ok == false {
                Err(WinError::from_win32())
            } else {
                Ok(())
            };

            match result.as_ref().map_err(|e| e.code()) {
                // Normally, for an async operation, a call like
                // `ReadFile` would return `false`, and the error code
                // would be `ERROR_IO_PENDING`.  But in some situations,
                // `ReadFile` can complete synchronously (returns `true`).
                // Even if it does, a notification that the IO completed
                // is still sent to the IO completion port that this
                // handle is part of, meaning that we don't have to do any
                // special handling for sync-completed operations.
                Ok(_) => Ok(()),
                Err(err) => {
                    if err == Foundation::ERROR_IO_PENDING.to_hresult() {
                        Ok(())
                    } else if err == Foundation::ERROR_BROKEN_PIPE.to_hresult() {
                        win32_trace!("[$ {:?}] BROKEN_PIPE straight from ReadFile", self.handle);

                        let async_data = self.r#async.take().unwrap().into_inner();
                        self.handle = async_data.handle;
                        self.read_buf = async_data.buf;

                        Err(WinIpcError::ChannelClosed)
                    } else {
                        let async_data = self.r#async.take().unwrap().into_inner();
                        self.handle = async_data.handle;
                        self.read_buf = async_data.buf;

                        result.map_err(|e| e.into())
                    }
                },
            }
        }
    }

    /// Called when we receive an IO Completion Packet for this handle.
    ///
    /// During its course, this method moves `async.buf` back into `read_buf`,
    /// thus making it accessible from normal code again;
    /// so `get_message()` can extract the received messages from the buffer.
    ///
    /// Invoking this is unsafe, since calling it in error
    /// while an async read is actually still in progress in the kernel
    /// would have catastrophic effects,
    /// as the `async` data is still mutably aliased by the kernel in that case!
    /// (See documentation of the `async` field.)
    ///
    /// Also, this method relies on `async` actually having valid data,
    /// i.e. nothing should modify its constituent fields
    /// between receiving the completion notification from the kernel
    /// and invoking this method.
    pub(super) unsafe fn notify_completion(
        &mut self,
        io_result: Result<(), WinError>,
    ) -> Result<(), WinIpcError> {
        win32_trace!(
            "[$ {:?}] notify_completion",
            self.r#async.as_ref().unwrap().alias().handle
        );

        // Regardless whether the kernel reported success or error,
        // it doesn't have an async read operation in flight at this point anymore.
        // (And it's safe again to access the `async` data.)
        let async_data = self.r#async.take().unwrap().into_inner();
        self.handle = async_data.handle;
        let ov = async_data.ov;
        self.read_buf = async_data.buf;

        match io_result {
            Ok(()) => {},
            Err(err) => {
                if err.code() == Foundation::ERROR_BROKEN_PIPE.to_hresult() {
                    // Remote end closed the channel.
                    return Err(WinIpcError::ChannelClosed);
                }
                return Err(err.into());
            },
        }

        let nbytes = ov.InternalHigh as u32;
        let offset = ov.Anonymous.Anonymous.Offset;

        assert!(offset == 0);

        let new_size = self.read_buf.len() + nbytes as usize;
        win32_trace!(
            "nbytes: {}, offset {}, buf len {}->{}, capacity {}",
            nbytes,
            offset,
            self.read_buf.len(),
            new_size,
            self.read_buf.capacity()
        );
        assert!(new_size <= self.read_buf.capacity());
        self.read_buf.set_len(new_size);

        Ok(())
    }

    /// Attempt to conclude an already issued async read operation.
    ///
    /// If successful, the result will be ready for picking up by `get_message()`.
    ///
    /// (`get_message()` might still yield nothing though,
    /// in case only part of the message was received in this read,
    /// and further read operations are necessary to get the rest.)
    ///
    /// In non-blocking mode, this may return with `WinError:NoData`,
    /// while the async operation remains in flight.
    /// The read buffer remains unavailable in that case,
    /// since it's still aliased by the kernel.
    /// (And there is nothing new to pick up anyway.)
    /// It will only become available again
    /// when `fetch_async_result()` returns successfully upon retry.
    /// (Or the async read is aborted with `cancel_io()`.)
    pub(super) fn fetch_async_result(&mut self, blocking_mode: BlockingMode) -> Result<(), WinIpcError> {
        unsafe {
            // Get the overlapped result, blocking if we need to.
            let mut nbytes: u32 = 0;
            let timeout = match blocking_mode {
                BlockingMode::Blocking => INFINITE,
                BlockingMode::Nonblocking => 0,
                BlockingMode::Timeout(duration) => {
                    duration.as_millis().try_into().unwrap_or(INFINITE)
                },
            };
            let ok = Win32IO::GetOverlappedResultEx(
                self.r#async.as_ref().unwrap().alias().handle.as_raw(),
                &mut ***self.r#async.as_mut().unwrap().alias_mut().ov.deref_mut(),
                &mut nbytes,
                timeout,
                false,
            );
            windows::Win32::System::Threading::ResetEvent(
                self.r#async
                    .as_mut()
                    .unwrap()
                    .alias_mut()
                    .ov
                    .deref_mut()
                    .hEvent,
            );
            let io_result = if ok == false {
                let err = Foundation::GetLastError();
                if blocking_mode != BlockingMode::Blocking && err == Foundation::ERROR_IO_INCOMPLETE {
                    // Async read hasn't completed yet.
                    // Inform the caller, while keeping the read in flight.
                    return Err(WinIpcError::NoData);
                }
                // Timeout has elapsed, so we must cancel the read operation before proceeding
                if err == Foundation::WAIT_TIMEOUT {
                    self.cancel_io();
                    return Err(WinIpcError::NoData);
                }
                // We pass err through to notify_completion so
                // that it can handle other errors.
                Err(WinError::new(err.to_hresult(), HSTRING::new()).into())
            } else {
                Ok(())
            };

            // Notify that the read completed, which will update the
            // read pointers
            self.notify_completion(io_result)
        }
    }

    pub(super) fn get_message(
        &mut self,
    ) -> Result<Option<(Vec<u8>, Vec<OsOpaqueIpcChannel>, Vec<OsIpcSharedMemory>)>, WinIpcError>
    {
        // Never touch the buffer while it's still mutably aliased by the kernel!
        if self.r#async.is_some() {
            return Ok(None);
        }

        let drain_bytes;
        let result;
        if let Some(message) = Message::from_bytes(&self.read_buf) {
            let mut channels: Vec<OsOpaqueIpcChannel> = vec![];
            let mut shmems: Vec<OsIpcSharedMemory> = vec![];
            let mut big_data = None;

            if let Some(oob) = message.oob_data() {
                win32_trace!("[$ {:?}] msg with total {} bytes, {} channels, {} shmems, big data handle {:?}",
                     self.handle, message.data_len, oob.channel_handles.len(), oob.shmem_handles.len(),
                     oob.big_data_receiver_handle);

                for handle in oob.channel_handles {
                    channels.push(OsOpaqueIpcChannel::new(WinHandle::new(Foundation::HANDLE(handle))));
                }

                for (handle, size) in oob.shmem_handles {
                    shmems.push(
                        OsIpcSharedMemory::from_handle(
                            WinHandle::new(Foundation::HANDLE(handle)),
                            size as usize,
                        )
                        .unwrap(),
                    );
                }

                if oob.big_data_receiver_handle.is_some() {
                    let (handle, big_data_size) = oob.big_data_receiver_handle.unwrap();
                    let receiver = OsIpcReceiver::from_handle(WinHandle::new(Foundation::HANDLE(handle)));
                    big_data = Some(receiver.recv_raw(big_data_size as usize)?);
                }
            }

            let buf_data = big_data.unwrap_or_else(|| message.data().to_vec());


            win32_trace!(
                "[$ {:?}] get_message success -> {} bytes, {} channels, {} shmems",
                self.handle,
                buf_data.len(),
                channels.len(),
                shmems.len()
            );
            drain_bytes = Some(message.size());
            result = Some((buf_data, channels, shmems));
        } else {
            drain_bytes = None;
            result = None;
        }

        if let Some(size) = drain_bytes {
            // If the only valid bytes in the buffer are what we just
            // consumed, then just set the vector's length to 0.  This
            // avoids reallocations as in the drain() case, and is
            // a significant speedup.
            if self.read_buf.len() == size {
                self.read_buf.clear();
            } else {
                self.read_buf.drain(0..size);
            }
        }

        Ok(result)
    }

    pub(super) fn add_to_iocp(&mut self, iocp: &WinHandle, entry_id: u64) -> Result<(), WinIpcError> {
        unsafe {
            assert!(self.entry_id.is_none());

            let completion_key = self.handle.as_raw().0;
            Win32IO::CreateIoCompletionPort(
                self.handle.as_raw(),
                iocp.as_raw(),
                completion_key as usize,
                0,
            )?;
        }

        self.entry_id = Some(entry_id);

        // The readers in the IOCP need to have async reads in flight,
        // so they can actually get completion events --
        // otherwise, a subsequent `select()` call would just hang indefinitely.
        self.start_read()
    }

    /// Specialized read for out-of-band data ports.
    ///
    /// Here the buffer size is known in advance,
    /// and the transfer doesn't have our typical message framing.
    ///
    /// It's only valid to call this as the one and only call after creating a MessageReader.
    pub(super) fn read_raw_sized(mut self, size: usize) -> Result<Vec<u8>, WinIpcError> {
        assert!(self.read_buf.len() == 0);

        self.read_buf.reserve(size);
        while self.read_buf.len() < size {
            // Because our handle is asynchronous, we have to do a two-part read --
            // first issue the operation, then wait for its completion.
            match self.start_read() {
                Err(WinIpcError::ChannelClosed) => {
                    // If the helper channel closes unexpectedly
                    // (i.e. before supplying the expected amount of data),
                    // don't report that as a "sender closed" condition on the main channel:
                    // rather, fail with the actual raw error code.
                    return Err(WinError::new(
                        Foundation::ERROR_BROKEN_PIPE.to_hresult(),
                        HSTRING::from("ReadFile"),
                    )
                    .into());
                },
                Err(err) => return Err(err),
                Ok(()) => {},
            };
            match self.fetch_async_result(BlockingMode::Blocking) {
                Err(WinIpcError::ChannelClosed) => {
                    return Err(WinError::new(
                        Foundation::ERROR_BROKEN_PIPE.to_hresult(),
                        HSTRING::from("ReadFile"),
                    )
                    .into());
                },
                Err(err) => return Err(err),
                Ok(()) => {},
            };
        }

        Ok(mem::replace(&mut self.read_buf, vec![]))
    }

    /// Get raw handle of the receive port.
    ///
    /// This is only for debug tracing purposes, and must not be used for anything else.
    pub(super) fn get_raw_handle(&self) -> Foundation::HANDLE {
        self.handle.as_raw()
    }
}
